package com.dataseek.iot.mqtt.config;

import com.alibaba.fastjson.JSONObject;
import com.dataseek.iot.mqtt.receiver.MqttMessageReceiver;
import com.dataseek.iot.mysql.entity.MqttTopic;
import com.dataseek.iot.mysql.service.IMqttTopicService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;
import org.springframework.http.*;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.Base64Utils;
import org.springframework.web.client.RestTemplate;

import java.nio.charset.StandardCharsets;
import java.util.*;

/**
 * @ClassName MqttOutboundConfiguration
 * @Description
 * @Author lcy
 * @Date 20:45 2019/7/25
 * @Version 1.0
 **/
@Configuration
@Slf4j
public class MqttBoundConfiguration {

    @Autowired
    MqttProperties mqttProperties;

    @Autowired
    MqttMessageReceiver mqttMessageReceiver;

    @Autowired
    RestTemplate restTemplate;


    private static MqttPahoMessageDrivenChannelAdapter adapter;


    @Value("${mqtt.username}")
    private String username;

    @Value("${mqtt.password}")
    private String password;

    @Value("${mqtt.prefix}")
    private String prefix;

    @Value("${mqtt.host}")
    private String host;

    ResponseEntity<JSONObject> restInit(String uri, HttpMethod method){
        HttpHeaders header = new HttpHeaders();
        String auth = username + ":" + password;
        byte[] authentication = auth.getBytes(StandardCharsets.UTF_8);
        byte[] base64Authentication = Base64Utils.encode(authentication);
        String baseCredential = new String(base64Authentication);
        header.add(HttpHeaders.AUTHORIZATION,  "Basic " + baseCredential);
        header.setAccept(Collections.singletonList(MediaType.APPLICATION_JSON));
        HttpEntity<JSONObject> request = new HttpEntity<JSONObject>(header);
        return restTemplate.exchange(uri, method, request, JSONObject.class);
    }

    /**
     * 添加topic
     * @param topics
     */
    public void addListenTopic(List<String> topics){
        if(adapter==null){
            adapter = new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getClientId(), mqttClientFactory(), "");
        }
        for(String topic : topics){
            adapter.addTopic(topic);
            adapter.setCompletionTimeout(5000);
            adapter.setQos(1);
            adapter.setOutputChannel(mqttInputChannel());
        }
        log.info("主题动态添加成功!");
    }

    /**
     * 移除topic
     * @param topics
     */
    public void removeListenTopic(List<String> topics){
        if(adapter==null){
            adapter = new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getClientId(), mqttClientFactory(), "");
        }
        for(String topic: topics){
            adapter.removeTopic(topic);
        }
    }


    @Bean
    public MqttConnectOptions getMqttConnectOptions() {
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        mqttConnectOptions.setServerURIs(mqttProperties.getBrokerUrl().split(","));
        mqttConnectOptions.setKeepAliveInterval(mqttProperties.getKeepAliveInterval());
        mqttConnectOptions.setMaxInflight(mqttProperties.getMaxInflight());
        mqttConnectOptions.setUserName(mqttProperties.getUsername());
        mqttConnectOptions.setPassword(mqttProperties.getPassword().toCharArray());
//        mqttConnectOptions.setWill(mqttProperties.getClientId()+"/offline", "offline".getBytes(), 2, true);
        mqttConnectOptions.setAutomaticReconnect(true);
        return mqttConnectOptions;
    }

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(getMqttConnectOptions());
        return factory;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler =
                new MqttPahoMessageHandler(mqttProperties.getClientId(), mqttClientFactory());
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic(mqttProperties.getTopic());
        messageHandler.setDefaultQos(0);
        return messageHandler;
    }

    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer channelInbound(MessageChannel mqttInputChannel, MqttPahoClientFactory factory) {
        //TODO 每次加载内存或者数据库中注册的topic
//        List<MqttTopic> storedTopics = mqttTopicService.list();
//
//        List<String> topicList  = new ArrayList();
//
//        String[] topics;
//        if(storedTopics.size()>0){
//            for(MqttTopic topic : storedTopics){
//                topicList.add(topic.getTopicCode());
//
//            }
//            topicList.add(prefix+host+"/clients/+/disconnected");
//            topicList.add(prefix+host+"/clients/+/connected");
//            topics = topicList.toArray(new String[topicList.size()]);
//
//        }else {
//            topics = mqttProperties.getTopic().split(",");
//        }
        adapter = new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getClientId(), factory,"#");
        adapter.addTopic("$SYS/brokers/+/clients/#");
        adapter.setCompletionTimeout(50000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel);
        return adapter;
    }

    /**
     * mqtt入站消息处理工具，对于指定消息入站通道接收到生产者生产的消息后处理消息的工具。
     *
     * @return
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler mqttMessageHandler() {
        return this.mqttMessageReceiver;
    }

//    public List<String> getEMQTopics(){
//        StringBuffer uri = new StringBuffer("http://")
//                .append("iot.capwill.net")
//                .append(":")
//                .append("8081")
//                .append("/api")
//                .append("/v4")
//                .append("/nodes/")
//                .append("capwill@127.0.0.1/")
//                .append("subscriptions")
//                .append("?")
//                .append("_page=1&_limit=1000");
//        ResponseEntity<JSONObject> result = restInit(uri.toString(), HttpMethod.GET);
//        JSONArray array = result.getBody().getJSONArray("data");
//        List<String> topics = new ArrayList<>();
//        for(Object o : array){
//            Map m  = (Map) o;
//            String topic = m.get("topic").toString();
//            topics.add(topic);
//        }
//        return topics;
//    }
}
